package com.sub.spark.core.rdd.demo;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import scala.Tuple2;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;

/**
 * @ClassName SubHotTopTen
 * @Description: 品类热榜：基于用户行为数据，分析计算热榜
 * 案例数据解释：每行数据为一条用户行为记录，数据以“_“分隔，每个数据项从左至右分别为：
 * 用户点击行为的日期_用户的ID_Session的ID_某个页面的ID_动作的时间点_用户搜索的关键词_点击某一个商品品类的ID_某一个商品的ID
 * 一次订单中所有品类的ID集合_一次订单中所有商品的ID集合_一次支付中所有品类的ID集合_一次支付中所有商品的ID集合_城市 id
 *
 * 行为分类：
 * 搜索行为：用户搜索的关键词不为null
 * 点击行为：点击某一个商品品类的ID不为-1
 * 下单行为：一次订单中所有品类的ID集合不为null
 * 支付行为：一次支付中所有品类的ID集合不为null
 *
 *  * 为了更好的泛用性，当前案例按照点击次数进行排序，如果点击相同，按照下单数，如果下单还是相同，按照支付数。
 *
 * Spark开发原则：多什么删什么，少什么补什么
 *
 * 扩展：使用kryo序列化，是java序列化性能的十倍。
 * @Author Submerge.
 * @Since 2025/5/20 14:21
 * @Version 1.0
 */
public class SubHotTopTen {

    public static void main(String[] args) {

        SparkConf conf = new SparkConf().setAppName("SubHotTopTen").setMaster("local[2]");

        JavaSparkContext javaSparkContext = new JavaSparkContext(conf);

        //1. 从文件创建RDD
        JavaRDD<String> textFileRDD = javaSparkContext.textFile("data/demo/spark/user_visit_action.txt");

        long totalCount = textFileRDD.count();
        System.out.println("初始总用户行为数据行："+ totalCount);


        //2.过滤多余数据：当前需求无需统计搜索行为，所以仅保留搜索关键词为空的数据统计
        JavaRDD<String> filterRDD = textFileRDD.filter(line -> {
            String[] lineArray = line.split("_");
            return "null".equals(lineArray[5]);
        });

        long staticsCount = filterRDD.count();
        System.out.println("过滤搜索行为数据行："+ (totalCount- staticsCount)+" ,剩余用户行为数据行: "+ staticsCount);

        //3.将用户行为数据行RDD转换用户对象RDD
        JavaRDD<UserVisitAction> mapRDD = filterRDD.map(line -> {
            String[] split = line.split("_");
            return new UserVisitAction(
                    split[0], split[1], split[2], split[3], split[4], split[5], split[6], split[7], split[8], split[9], split[10], split[11], split[12]
            );
        });

        System.out.println("After map RDD count is :" + mapRDD.count());

        //4.将用户对象扁平映射为品类数量对象RDD
        JavaRDD<CategoryCountInfo> flatMapRDD = mapRDD.flatMap(userVisitAction -> {
            ArrayList<CategoryCountInfo> countInfos = new ArrayList<>();
            if (!userVisitAction.getClick_category_id().equals("-1")) {
                // 当前为点击数据
                countInfos.add(new CategoryCountInfo(userVisitAction.getClick_category_id(), 1L, 0L, 0L));
            } else if (!userVisitAction.getOrder_category_ids().equals("null")) {
                // 当前为订单数据
                String[] orders = userVisitAction.getOrder_category_ids().split(",");
                for (String order : orders) {
                    countInfos.add(new CategoryCountInfo(order, 0L, 1L, 0L));
                }
            } else if (!userVisitAction.getPay_category_ids().equals("null")) {
                // 当前为支付数据
                String[] pays = userVisitAction.getPay_category_ids().split(",");
                for (String pay : pays) {
                    countInfos.add(new CategoryCountInfo(pay, 0L, 0L, 1L));
                }
            }
            return countInfos.iterator();
        });


        //5、对品类数量RDD 转换为KV类型RDD
        JavaPairRDD<String, CategoryCountInfo> mapToPairRDD = flatMapRDD.mapToPair(categoryCountInfo -> {
            return new Tuple2<>(categoryCountInfo.getCategoryId(), categoryCountInfo);
        });



        //6、使用ReduceByKey（对于相同的K，将V两两聚合） 分组聚合
        JavaPairRDD<String, CategoryCountInfo> reduceByKeyRDD = mapToPairRDD.reduceByKey((Function2<CategoryCountInfo, CategoryCountInfo, CategoryCountInfo>) (categoryCountInfo, categoryCountInfo2) -> new CategoryCountInfo(categoryCountInfo.getCategoryId(),
                categoryCountInfo.getClickCount() + categoryCountInfo2.getClickCount(),
                categoryCountInfo.getOrderCount() + categoryCountInfo2.getOrderCount(),
                categoryCountInfo.getPayCount() + categoryCountInfo2.getPayCount()));


        long groupCount = reduceByKeyRDD.count();
        System.out.println("按照品类分组聚合后记录数为："+groupCount);


        JavaRDD<CategoryCountInfo> categoryCountInfoJavaRDD = reduceByKeyRDD.map(Tuple2::_2);

        //7、排序
        JavaRDD<CategoryCountInfo> sortByRDD = categoryCountInfoJavaRDD.sortBy((Function<CategoryCountInfo, Object>) categoryCountInfo -> categoryCountInfo, false, 2);

        //8、取排序后的前十条数据
        List<CategoryCountInfo> takeTop10 = sortByRDD.take(10);

        //最后打印结果
        System.out.println("品类热榜TOP10计算结果如下：");
        takeTop10.forEach(System.out::println);

        //关闭
        javaSparkContext.stop();

    }
}


@NoArgsConstructor
@AllArgsConstructor
@Data
class UserVisitAction implements Serializable{

    private String date;
    private String user_id;
    private String session_id;
    private String page_id;
    private String action_time;
    private String search_keyword;
    private String click_category_id;
    private String click_product_id;
    private String order_category_ids;
    private String order_product_ids;
    private String pay_category_ids;
    private String pay_product_ids;
    private String city_id;

}

@Data
@NoArgsConstructor
@AllArgsConstructor
class CategoryCountInfo implements Serializable,Comparable<CategoryCountInfo>{

    private String categoryId;
    private Long clickCount;
    private Long orderCount;
    private Long payCount;


    @Override
    public int compareTo(CategoryCountInfo o) {
        // 小于返回-1,等于返回0,大于返回1
        if (this.getClickCount().equals(o.getClickCount())) {
            if (this.getOrderCount().equals(o.getOrderCount())) {
                if (this.getPayCount().equals(o.getPayCount())) {
                    return 0;
                } else {
                    return this.getPayCount() < o.getPayCount() ? -1 : 1;
                }
            } else {
                return this.getOrderCount() < o.getOrderCount() ? -1 : 1;
            }
        } else {
            return this.getClickCount() < o.getClickCount() ? -1 : 1;
        }
    }

    @Override
    public String toString() {
        return "CategoryCountInfo{" +
                "categoryId='" + categoryId + '\'' +
                ", clickCount=" + clickCount +
                ", orderCount=" + orderCount +
                ", payCount=" + payCount +
                '}';
    }
}
